package com.mgface.zookeeper.premitive;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class 数据发布订阅_配置管理 {
	private static Logger log = Logger.getLogger(数据发布订阅_配置管理.class);
	private static ZooKeeper client;
	private static String path = "/ps1";
	public static void main(String[] args) throws Exception {
		CountDownLatch on = new CountDownLatch(1);
		client = new ZooKeeper("localhost:2181,localhost:2182,localhost:2183", 10_000, e -> {
			try {
				if (e.getState() == KeeperState.SyncConnected) {
					switch (e.getType()) {
					case None:
						if (e.getPath() == null) {
							on.countDown();
						}
						break;
					case NodeCreated:
						Stat stat = new Stat();
						byte[] v = client.getData(path, true, stat);// 获取数据，并且监听path节点的变动
						log.info("节点创建成功,值:" + new String(v));
						break;
					case NodeDeleted:
						log.info("节点:" + path + "被删除");
						break;
					case NodeDataChanged:
						Stat stat2 = new Stat();
						byte[] v2 = client.getData(path, true, stat2);
						log.info("新值:" + new String(v2));
						break;
					case NodeChildrenChanged:
						List<String> list = client.getChildren(path, true);// 获取子节点的相对路径，并且监听子节点的列表变化
						log.info("节点变化:" + list);
						break;
					default:
						break;
					}
				}
			} catch (Exception error) {
				error.getMessage();
			}

		});
		log.info("ZK状态:" + client.getState().toString());
		on.await();
		log.info("zookeeper session 建立完成");
		Stat stat = client.exists(path, true);// 给当前节点设置监控
		log.info("根节点("+path+"):" + (stat == null ? "不存在" : "存在"));
		if(stat==null){
			client.create(path, "init".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			log.info("success create znode:" + path);
		}
		//创建发布节点
		String v_path = path+"/v1";
		Stat publish = client.exists(v_path, true);
		if(publish==null){
			path = v_path;
			client.create(path, "{database:mysql,user:admin,password:admin}".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
			log.info("创建子节点:"+path);
		}
		TimeUnit.SECONDS.sleep(5);
		
		CountDownLatch cdl = new CountDownLatch(1);
		//创建三个线程来订阅信息
		Watcher watcher = e->{
			try {
				if (e.getState() == KeeperState.SyncConnected) {
					switch (e.getType()) {
					case None:
						if (e.getPath() == null) {
							log.info("当前线程:"+Thread.currentThread().getName()+"连接成功...");
						}
						break;
					case NodeDataChanged:
						Stat stat2 = new Stat();
						byte[] v2 = client.getData(path, true, stat2);
						log.info("当前线程:"+Thread.currentThread().getName()+"新值:" + new String(v2));
						break;
					default:
						break;
					}
				}
			} catch (Exception error) {
				error.getMessage();
			}
		};
		ExecutorService exec = Executors.newFixedThreadPool(4);
		for (int i = 0; i < 4; i++) {
			exec.submit(()->{
				try {
					ZooKeeper zk = new  ZooKeeper("localhost:2181,localhost:2182,localhost:2183", 10_000,watcher);
					Stat st = new Stat();
					byte[] bytes = zk.getData(path, true,st);//设置path的监控
					log.info("当前线程:"+Thread.currentThread().getName()+"获取的值:"+new String(bytes));
					cdl.await();
				} catch (Exception e1) {
					e1.printStackTrace();
				}
			});
		}
		exec.shutdown();
		cdl.countDown();
		TimeUnit.SECONDS.sleep(5);
		System.out.println("过去了5S执行发布新数据");
		client.setData(path, "{database:oracle,user:admin,password:admin}".getBytes(), -1);
		TimeUnit.SECONDS.sleep(500);
	}

}
